package com.atcyj.gamll.realtime.common.util;

import com.atcyj.gamll.realtime.common.constant.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.kafka.clients.producer.ProducerConfig;



/**
 * @author: cyj
 * @date: 2024/8/16
 */
public class FlinkSinkUtil {
    public static KafkaSink<String> getKafkaSink(String topic) {

        return KafkaSink.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(topic)
                                // new SimpleStringSchema() 发数据时，不能发null值，我们已经把空值都过滤掉了
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build())
                .setTransactionalIdPrefix("atcyj-" + topic + System.currentTimeMillis())
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) //精准一次性
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "") //事务超时时间
                .build();

    }
}
